package rpc.core.plugin.impl;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.core.env.Environment;
import org.springframework.util.CollectionUtils;
import rpc.core.FlowMap;
import rpc.core.RpcApp;
import rpc.core.annotations.Flow;
import rpc.core.plugin.ServicePlugin;
import rpc.core.plugin.impl.proxy.AsyncFlowService;
import rpc.core.plugin.impl.proxy.FlowServiceProxy;
import rpc.core.plugin.impl.proxy.ServiceProxy;
import rpc.utils.ReflectionUtils;

public class FlowServicePlugin implements ServicePlugin {

  private static final Logger logger = LoggerFactory.getLogger(FlowServicePlugin.class);
  private final ApplicationContext context;
  private final Environment environment;
  private final Map<String, ServiceProxy> flowProxyMap = new HashMap<>();
  private int queueSize = 1000;
  private int threads = 0;
  private int maxThreads = 0;
  private ThreadPoolExecutor executor = null;

  public FlowServicePlugin(ApplicationContext context, Environment environment) {
    this.context = context;
    this.environment = environment;
  }

  @Override
  public void init() {
    List<String> classNames = getClassNameList(this.environment);
    if (CollectionUtils.isEmpty(classNames)) {
      return;
    }
    List<? extends Class<?>> classList = classNames.stream().map(ReflectionUtils::getClass).filter(c -> Objects.nonNull(c.getAnnotation(Flow.class)))
        .collect(Collectors.toList());
    threads = intConfig(environment, "rpc.flow.threads", 8);
    maxThreads = intConfig(environment, "rpc.flow.maxthreads", 8);
    queueSize = intConfig(environment, "rpc.flow.queue", 10000);
    this.executor = new ThreadPoolExecutor(threads, maxThreads, 0, TimeUnit.MINUTES, new ArrayBlockingQueue<>(queueSize)
        , new ThreadFactoryBuilder().setNameFormat("pool_flow_%d").build());
    classList.forEach(flowClazz -> {
      Flow flow = flowClazz.getAnnotation(Flow.class);
      flowProxyMap.put(flow.value().toLowerCase(), new FlowServiceProxy().setContext(context).setEnvironment(environment)
          .setFlowClass(flowClazz).setFlowName(flow.value()).setExecutorService(executor)
          .setBootstrap(context.getBean(RpcApp.class).getBootstrap()).setAsyncFlowService(context.getBean(AsyncFlowService.class))
          .init());
    });
    logger.info("flowProxyMap:{}", String.join(",", flowProxyMap.keySet()));
    executor.prestartAllCoreThreads();
  }

  @Override
  public java.util.Set<String> getFlowName() {
    return this.flowProxyMap.keySet();
  }

  @Override
  public void close() {
    if (Objects.nonNull(executor)) {
      this.executor.shutdown();
    }
  }

  @Override
  public Map<String, ServiceProxy> getFlowServiceProxy() {
    return this.flowProxyMap;
  }

  @Override
  public Map<String, Object> dump() {
    FlowMap flowMap = new FlowMap();
    flowMap.append("threads", threads);
    flowMap.append("maxThreads", maxThreads);
    return flowMap.toMap();
  }
}